package com.roc.rocketmqconsumer.messageType;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.common.message.MessageExt;

import java.util.List;

public class AsyncMsgConsumer {
    public static void main(String[] args) throws Exception {
        // 1、实例化消费者，指定消费者组名
        DefaultMQPushConsumer consumer = new DefaultMQPushConsumer("roc_consumer_group");
        // 2、设置NameServer的地址
        consumer.setNamesrvAddr("127.0.0.1:9876");
        // 3、订阅一个或者多个Topic，以及Tag来过滤需要消费的消息
        consumer.subscribe("TOPIC_ASYNC", "*");
        // 4、注册回调实现类来处理从broker拉取回来的消息
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs,
                                                            ConsumeConcurrentlyContext context) {
                for (MessageExt msg : msgs) {
                    try {
                        // 输出消息内容，这里简单打印消息体（需注意编码，示例假设是 UTF-8）
                        String msgBody = new String(msg.getBody(), "UTF-8");
                        System.out.println("接收消息：topic=" + msg.getTopic() + ", tag=" + msg.getTags() + ", 消息内容=" + msgBody);
                    } catch (Exception e) {
                        e.printStackTrace();
                        // 若消息处理异常，可根据情况决定是否重试等，这里返回 RECONSUME_LATER 表示稍后重试
                        return ConsumeConcurrentlyStatus.RECONSUME_LATER;
                    }
                }
                // 消息处理成功，返回 CONSUME_SUCCESS
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        // 启动消费者实例
        consumer.start();
        System.out.println("异步消息消费者已启动……");
    }
}